package com.tanhua.dubbo.service;

import cn.hutool.json.JSONUtil;
import com.tanhua.dubbo.pojo.Publish;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.util.List;

public class ConsumerService {

    @Value("${spring.mq.quanzi.topic.address}")
    private String address;

    @Autowired
    private TimeLineService timeLineService;

    public  void consumer(){


        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
            //2.设置命名服务器地址和端口
            consumer.setNamesrvAddr(address);
            //3.设置订阅的主题
            consumer.subscribe("publish","*");
            //4.编写监听,当topic1下有内容时获取对应的消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                // 监听同步消息状态
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        String s = new String(msg.getBody());
                        Publish publish = JSONUtil.toBean(s, Publish.class);


                        timeLineService.saveTimeLine(publish);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            //5.启动消费者
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

}
